package com.ustc.hewei.nettyrpc2_0.netty.server.net;

import com.ustc.hewei.nettyrpc2_0.common.codec.CommonDecode;
import com.ustc.hewei.nettyrpc2_0.common.codec.CommonEncode;
import com.ustc.hewei.nettyrpc2_0.common.exception.RpcException;
import com.ustc.hewei.nettyrpc2_0.common.factory.ThreadPoolFactory;
import com.ustc.hewei.nettyrpc2_0.common.hook.ShutdownHook;
import com.ustc.hewei.nettyrpc2_0.common.warp.RpcRequest;
import com.ustc.hewei.nettyrpc2_0.common.warp.RpcResponse;
import com.ustc.hewei.nettyrpc2_0.netty.server.handler.RequestHandler;
import com.ustc.hewei.nettyrpc2_0.zookeeper.register.ServiceRegister;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.timeout.IdleState;
import io.netty.handler.timeout.IdleStateEvent;
import io.netty.handler.timeout.IdleStateHandler;
import io.netty.util.ReferenceCountUtil;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;

import javax.annotation.Resource;
import java.io.PrintWriter;
import java.io.StringWriter;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.TimeUnit;

/**
* @description: 服务端主类
* @author hewei
* @date 2022/9/16 20:07
* @version 1.0
*/

@Slf4j
public class NettyServer extends RpcServer {
    private final RequestHandler requestHandler;
    private Thread thread;
    private final ExecutorService executorService;
    private final ServiceRegister serviceRegister;
    // TODO: 2022/10/28 可以设置一个服务的计数器，设置一个线程定时清零计数器，
    //  如果一段时间内服务调用超过指定次数，则返回默认信息并休息一段时间，或是采用redis加lua脚本的方式

    public NettyServer(int port, RequestHandler requestHandler, ServiceRegister serviceRegister) {
        super(port);
        this.requestHandler = requestHandler;
        this.serviceRegister = serviceRegister;
        executorService = ThreadPoolFactory.createDefaultThreadPool("requestHandlerPool");
    }

    @Override
    public void start() {
        ShutdownHook.getShutdownHook().addShutdownHook();
        thread = new Thread(() -> {
            NioEventLoopGroup boosGroup = new NioEventLoopGroup();
            NioEventLoopGroup workGroup = new NioEventLoopGroup();
            try {
                ServerBootstrap bootstrap = new ServerBootstrap();
                bootstrap.group(boosGroup, workGroup).channel(NioServerSocketChannel.class)
                        .option(ChannelOption.SO_BACKLOG, 128)
                        .childOption(ChannelOption.SO_KEEPALIVE, true)
                        .childHandler(new ChannelInitializer<SocketChannel>() {
                            @Override
                            protected void initChannel(SocketChannel socketChannel) throws Exception {
                                ChannelPipeline pipeline = socketChannel.pipeline();
                                pipeline.addLast(new IdleStateHandler(10, 0, 0, TimeUnit.SECONDS))
                                        .addLast(new CommonDecode())
                                        .addLast(new CommonEncode())
                                        .addLast(new ServerHandler());
                            }
                        });
                ChannelFuture channelFuture = bootstrap.bind(port).sync();
                channelFuture.channel().closeFuture().sync();
            } catch (InterruptedException e) {
                log.error("远程服务 {} 停止", port);
            } finally {
                workGroup.shutdownGracefully();
                boosGroup.shutdownGracefully();
            }
        });
        thread.start();
    }

    @Override
    public void stop() {
        if (thread != null && thread.isAlive()) {
            thread.interrupt();
        }
    }

    private class ServerHandler extends SimpleChannelInboundHandler<RpcRequest> {
        @Override
        protected void channelRead0(ChannelHandlerContext ctx, RpcRequest request) throws Exception {
            if (request.getHeartBeat()) {
                //log.info("收到客户端心跳");
                return;
            }
            executorService.submit(() -> {
                RpcResponse response = null;
                try {
                    Object data = requestHandler.handler(request);
                    response = RpcResponse.success(request, data);
                } catch (RpcException e) {
                    response = RpcResponse.fail(request, e.getMessage());
                }catch (Exception e) {
                    // TODO: 2022/9/25 使用AOP和filter实现统一异常处理
                    response = RpcResponse.fail(request, getStackTrace(e));
                    e.printStackTrace();
                } finally {
                    if (ctx.channel().isActive() && ctx.channel().isWritable()) {
                        RpcResponse finalResponse = response;
                        ctx.writeAndFlush(response).addListener(new ChannelFutureListener() {
                            @Override
                            public void operationComplete(ChannelFuture channelFuture) throws Exception {
                                log.info("发出请求 {}", request.getId());
                                if (!finalResponse.getIsSuccess()) {
                                    String serviceName = request.getInterfaceName();
                                    log.error("请求处理失败，删除服务 {}", serviceName);
                                    // 代表在服务端这边反射调用出现异常，对于这种错误，采取直接关闭此项服务的做法
                                    // 在发送成功之后再进行判断，因为先一步关闭服务，可能会导致消费端的监听器产生响应，直接关闭连接，导致异常信息没来得及传到消费端
                                    serviceRegister.unregister(serviceName);
                                }
                            }
                        });
                    } else {
                        log.error("通道不可写");
                    }
                    ReferenceCountUtil.release(request);
                }
            });
        }

        public String getStackTrace(Throwable ex) {
            StringWriter buf = new StringWriter();
            ex.printStackTrace(new PrintWriter(buf));

            return buf.toString();
        }

        @Override
        public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
            if (evt instanceof IdleStateEvent) {
                IdleState state = ((IdleStateEvent) evt).state();
                if (state == IdleState.READER_IDLE) {
                    // TODO: 2022/9/23 对于心跳超时的处理，
                    //  考虑将该handler对应的所有服务在zookeeper中的权重暂时调整为0，
                    //  通过监听器就会实时更新客户端的缓存，
                    //  问题是该怎么判断该handler已经恢复了呢？
                    log.info("长时间未收到心跳，关闭客户端 {}", ctx.channel().remoteAddress());
                    ctx.close();
                }
            } else {
                super.userEventTriggered(ctx, evt);
            }
        }

        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
            super.exceptionCaught(ctx, cause);
            throw new RuntimeException("服务端处理器异常", cause.getCause());
        }
    }
}
